package com.rongmei.jdsync.service.impl;

import com.rongmei.jdsync.client.MallGoodsFeignApi;
import com.rongmei.jdsync.config.PriceSyncConfig;
import com.rongmei.jdsync.dto.GoodsDetailDto;
import com.rongmei.jdsync.dto.GoodsDetailInfoResponseDto;
import com.rongmei.jdsync.dto.PriceChangeDto;
import com.rongmei.jdsync.dto.RBase;
import com.rongmei.jdsync.dto.sync.BatchProcessResult;
import com.rongmei.jdsync.dto.sync.PriceSyncResult;
import com.rongmei.jdsync.dto.sync.PushResult;
import com.rongmei.jdsync.dto.sync.RetryResult;
import com.rongmei.jdsync.dto.sync.SyncOptions;
import com.rongmei.jdsync.entity.Goods;
import com.rongmei.jdsync.entity.PriceSyncLog;
import com.rongmei.jdsync.repository.GoodsRepository;
import com.rongmei.jdsync.repository.PriceSyncLogRepository;
import com.rongmei.jdsync.service.GoodsService;
import com.rongmei.jdsync.service.JdCategorySyncService;
import com.rongmei.jdsync.service.PriceSyncService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 价格同步服务实现类
 * 提供商品价格同步的核心业务逻辑实现
 *
 * @author dongjun
 */
@Slf4j
@Service
public class PriceSyncServiceImpl implements PriceSyncService {

    private static final String SYNC_BATCH_PREFIX = "SYNC_";
    private static final int DEFAULT_BATCH_SIZE = 100;

    private final MallGoodsFeignApi mallGoodsFeignApi;
    private final GoodsRepository goodsRepository;
    private final PriceSyncLogRepository priceSyncLogRepository;
    private final JdCategorySyncService jdCategorySyncService;
    private final PriceSyncConfig priceSyncConfig;

    private final GoodsService goodsService;

    // 自引用，用于调用事务方法
    @Autowired
    private PriceSyncService self;



    /**
     * 构造函数注入依赖
     */
    public PriceSyncServiceImpl(MallGoodsFeignApi mallGoodsFeignApi,
                                GoodsRepository goodsRepository,
                                PriceSyncLogRepository priceSyncLogRepository,
                                JdCategorySyncService jdCategorySyncService,
                                PriceSyncConfig priceSyncConfig, GoodsService goodsService) {
        this.mallGoodsFeignApi = mallGoodsFeignApi;
        this.goodsRepository = goodsRepository;
        this.priceSyncLogRepository = priceSyncLogRepository;
        this.jdCategorySyncService = jdCategorySyncService;
        this.priceSyncConfig = priceSyncConfig;
        this.goodsService = goodsService;
    }

    @Override
    public PriceSyncResult executePriceSync() {
        SyncOptions defaultOptions = SyncOptions.builder()
                .priceThreshold(priceSyncConfig.getPriceChangeThreshold())
                .batchSize(DEFAULT_BATCH_SIZE)
                .build();
        return executePriceSync(defaultOptions);
    }

    @Override
    public PriceSyncResult executePriceSync(SyncOptions syncOptions) {
        long startTime = System.currentTimeMillis();
        String syncBatchId = generateSyncBatchId();
        log.info("开始执行价格同步任务，批次ID: {}, 配置: 阈值={}, 批量大小={}", syncBatchId, syncOptions.getPriceThreshold(), Integer.valueOf(syncOptions.getBatchSize()));
        try {
            // 检查是否启用价格同步
            if (!priceSyncConfig.isEnabled()) {
                log.warn("价格同步功能已禁用，跳过执行");
                return createEmptyResult(syncBatchId, System.currentTimeMillis() - startTime);
            }

            // 使用分批处理模式，避免内存溢出
            return executePriceSyncInBatches(syncBatchId, syncOptions, startTime);

        } catch (Exception e) {
            log.error("价格同步任务执行失败，批次ID: {}", syncBatchId, e);
            // 记录失败日志
            recordSyncFailure(syncBatchId, e.getMessage());
            throw new RuntimeException("价格同步任务执行失败: " + e.getMessage(), e);
        }
    }






    /**
     * 分批执行价格同步，避免内存溢出
     */
    private PriceSyncResult executePriceSyncInBatches(String syncBatchId, SyncOptions syncOptions, long startTime) {
        int totalSyncGoods = 0;
        int totalLocalFound = 0;
        int totalPriceChanged = 0;
        int totalPushSuccess = 0;
        int totalPushFailed = 0;
        BigDecimal totalChangeAmount = BigDecimal.ZERO;

        int page = 1;
        int pageSize = syncOptions != null ? syncOptions.getBatchSize() : DEFAULT_BATCH_SIZE;
        boolean hasMore = true;

        log.info("开始分批执行价格同步，页面大小: {}", Integer.valueOf(pageSize));

        while (hasMore) {
            try {
                log.info("处理第 {} 页数据", Integer.valueOf(page));

                // 1. 获取当前页的商品ID列表
                RBase<Long> response = mallGoodsFeignApi.getSyncGoodsPage(Integer.valueOf(page), Integer.valueOf(pageSize));
                //========自测 数据
//                List<Goods>  goodsList =  goodsService.findGoodsByPage(page,pageSize);
//                RBase<Long> response = new RBase<>();
//                response.setList(goodsList.stream().map(Goods::getGoodsId).collect(Collectors.toList()));
//                response.setStatus(Integer.valueOf(200));
//                response.setMsg("success");
                //========自测 数据

                if (response.getStatus() != 200 || response.getList() == null || response.getList().isEmpty()) {
                    log.warn("第 {} 页没有获取到数据，结束处理", Integer.valueOf(page));
                    break;
                }

                List<Long> pageGoodsIds = response.getList();
                totalSyncGoods += pageGoodsIds.size();

                log.debug("第 {} 页获取到 {} 个商品ID", Integer.valueOf(page), Integer.valueOf(pageGoodsIds.size()));

                // 2. 处理当前页的数据（每页一个事务）
                BatchProcessResult batchResult;
                try {
                    batchResult = self.processBatch(pageGoodsIds, syncBatchId, syncOptions);

                    // 3. 累计统计结果
                    totalLocalFound += batchResult.getLocalFoundCount();
                    totalPriceChanged += batchResult.getPriceChangedCount();
                    totalPushSuccess += batchResult.getPushSuccessCount();
                    totalPushFailed += batchResult.getPushFailedCount();
                    totalChangeAmount = totalChangeAmount.add(batchResult.getChangeAmount());

                    log.info("第 {} 页处理完成: 本地找到={}, 价格变更={}, 推送成功={}, 推送失败={}",
                            Integer.valueOf(page), Integer.valueOf(batchResult.getLocalFoundCount()), Integer.valueOf(batchResult.getPriceChangedCount()),
                            Integer.valueOf(batchResult.getPushSuccessCount()), Integer.valueOf(batchResult.getPushFailedCount()));

                } catch (Exception e) {
                    // 单个批次事务失败，记录错误但继续处理下一批次
                    log.error("第 {} 页处理失败，事务已回滚，继续处理下一页: {}", Integer.valueOf(page), e.getMessage());
                    // 将整页标记为失败
                    totalPushFailed += pageGoodsIds.size();
                }

                // 4. 判断是否还有下一页
                if (pageGoodsIds.size() < pageSize) {
                    log.info("第 {} 页返回数据量({})小于页面大小({})，处理完成", Integer.valueOf(page), Integer.valueOf(pageGoodsIds.size()), Integer.valueOf(pageSize));
                    hasMore = false;
                } else {
                    page++;

                    // 每处理几页后进行垃圾回收
                    if (page % 5 == 0) {
                        System.gc();
                        log.debug("已处理 {} 页，建议进行垃圾回收", Integer.valueOf(page - 1));
                    }
                    // 添加短暂延迟，避免对系统造成过大压力
                    Thread.sleep(50);
                }

            } catch (Exception e) {
                log.error("处理第 {} 页数据时发生异常", Integer.valueOf(page), e);
                // 假设整页都失败了
                totalPushFailed += pageSize;
                page++;
            }
        }
        // 生成最终结果
        long executionTime = System.currentTimeMillis() - startTime;
        PriceSyncResult result;

        if (totalPriceChanged == 0) {
            result = PriceSyncResult.noChange(syncBatchId, totalSyncGoods, totalLocalFound, executionTime);
        } else if (totalPushFailed == 0) {
            result = PriceSyncResult.success(syncBatchId, totalSyncGoods, totalLocalFound, totalPriceChanged, totalPushSuccess, executionTime);
        } else {
            result = PriceSyncResult.partialSuccess(syncBatchId, totalSyncGoods, totalLocalFound,
                                                  totalPriceChanged, totalPushSuccess, totalPushFailed, executionTime);
        }

        result.setTotalChangeAmount(totalChangeAmount);
        result.setSyncOptions(syncOptions);

        log.info("分批价格同步完成: {}", result);
        return result;
    }

    /**
     * 处理单个批次的数据
     * 每个批次作为一个独立的事务，确保数据一致性
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public BatchProcessResult processBatch(List<Long> goodsIds, String syncBatchId, SyncOptions syncOptions) {
        long batchStartTime = System.currentTimeMillis();
        log.info("开始处理批次数据，商品数量: {}, 批次ID: {}", Integer.valueOf(goodsIds.size()), syncBatchId);

        try {
            // 1. 查询本地商品信息
            List<Goods> localGoods = getLocalGoodsByIds(goodsIds);
            log.debug("本地找到商品数量: {}", Integer.valueOf(localGoods.size()));

            if (localGoods.isEmpty()) {
                log.debug("当前批次没有找到本地商品，跳过处理");
                return BatchProcessResult.empty();
            }
            // 2. 获取京东商品信息（注意：这个操作涉及外部API调用，在事务内执行需要控制时间）
            Map<Long, Goods> jdGoodsMap = getJdGoodsMap(goodsIds);
            log.debug("从京东获取商品数量: {}", Integer.valueOf(jdGoodsMap.size()));

            // 3. 比较价格差异
            List<PriceChangeDto> priceChanges = comparePrices(localGoods, jdGoodsMap, syncOptions.getPriceThreshold());
            log.debug("发现价格变更商品数量: {}", Integer.valueOf(priceChanges.size()));

            if (priceChanges.isEmpty()) {
                log.debug("当前批次没有价格变更，事务正常提交");
                return BatchProcessResult.noChange(localGoods.size());
            }

            // 4. 记录同步日志
            List<PriceSyncLog> syncLogs = createSyncLogs(priceChanges, syncBatchId);
            priceSyncLogRepository.saveAll(syncLogs);
            log.debug("已保存 {} 条同步日志", Integer.valueOf(syncLogs.size()));

            // 5. 推送价格变更（这个操作可能失败，但不影响事务）
            PushResult pushResult = pushPriceChanges(priceChanges, syncBatchId);
            log.debug("推送结果: 成功={}, 失败={}", Integer.valueOf(pushResult.getSuccessCount()), Integer.valueOf(pushResult.getFailedCount()));

            // 6. 更新本地价格（只更新推送成功的，在事务内）
            if (pushResult.getSuccessCount() > 0) {
                updateLocalPrices(pushResult.getSuccessChanges());
                log.debug("已更新 {} 个商品的本地价格", Integer.valueOf(pushResult.getSuccessCount()));
            }

            // 7. 计算变更金额
            BigDecimal changeAmount = priceChanges.stream()
                    .map(PriceChangeDto::getChangeAmount)
                    .filter(Objects::nonNull)
                    .reduce(BigDecimal.ZERO, BigDecimal::add);

            long batchEndTime = System.currentTimeMillis();
            log.info("批次处理完成，耗时: {}ms, 本地商品: {}, 价格变更: {}, 推送成功: {}, 推送失败: {}",
                    Long.valueOf(batchEndTime - batchStartTime), Integer.valueOf(localGoods.size()), Integer.valueOf(priceChanges.size()),
                    Integer.valueOf(pushResult.getSuccessCount()), Integer.valueOf(pushResult.getFailedCount()));

            return BatchProcessResult.builder()
                    .localFoundCount(localGoods.size())
                    .priceChangedCount(priceChanges.size())
                    .pushSuccessCount(pushResult.getSuccessCount())
                    .pushFailedCount(pushResult.getFailedCount())
                    .changeAmount(changeAmount)
                    .build();

        } catch (Exception e) {
            log.error("处理批次数据时发生异常，商品数量: {}, 批次ID: {}, 事务将回滚", Integer.valueOf(goodsIds.size()), syncBatchId, e);
            // 事务会自动回滚，返回失败结果
            throw new RuntimeException("批次处理失败: " + e.getMessage(), e);
        }
    }



    @Override
    public List<Goods> getLocalGoodsByIds(List<Long> goodsIds) {
        if (goodsIds == null) {
            throw new IllegalArgumentException("商品ID列表不能为null");
        }

        if (goodsIds.isEmpty()) {
            log.debug("商品ID列表为空，返回空结果");
            return Collections.emptyList();
        }

        log.debug("开始批量查询本地商品信息，总数量: {}", Integer.valueOf(goodsIds.size()));

        // 分批查询，避免IN查询参数过多
        List<Goods> allGoods = new ArrayList<>();
        int batchSize = DEFAULT_BATCH_SIZE;
        int totalBatches = (int) Math.ceil((double) goodsIds.size() / batchSize);

        for (int i = 0; i < goodsIds.size(); i += batchSize) {
            int endIndex = Math.min(i + batchSize, goodsIds.size());
            List<Long> batchIds = goodsIds.subList(i, endIndex);

            try {
                List<Goods> batchGoods = goodsRepository.findByGoodsIdIn(batchIds);
                allGoods.addAll(batchGoods);

                int currentBatch = (i / batchSize) + 1;
                log.debug("完成第 {}/{} 批查询，本批查询 {} 个商品，找到 {} 个",
                         Integer.valueOf(currentBatch), Integer.valueOf(totalBatches), Integer.valueOf(batchIds.size()), Integer.valueOf(batchGoods.size()));

            } catch (Exception e) {
                log.error("批量查询商品信息失败，批次: {}-{}", Integer.valueOf(i), Integer.valueOf(endIndex), e);
                // 继续处理下一批，不中断整个流程
            }
        }

        log.info("本地商品查询完成，请求 {} 个，找到 {} 个", Integer.valueOf(goodsIds.size()), Integer.valueOf(allGoods.size()));

        // 记录未找到的商品ID（用于调试）
        if (allGoods.size() < goodsIds.size()) {
            Set<Long> foundIds = allGoods.stream().map(Goods::getGoodsId).collect(Collectors.toSet());
            List<Long> notFoundIds = goodsIds.stream()
                    .filter(id -> !foundIds.contains(id))
                    .collect(Collectors.toList());
            log.debug("未在本地找到的商品ID数量: {}", Integer.valueOf(notFoundIds.size()));
        }

        return allGoods;
    }

    @Override
    public List<PriceChangeDto> comparePrices(List<Goods> localGoods, Map<Long, Goods> jdGoodsMap, BigDecimal threshold) {
        if (localGoods == null || localGoods.isEmpty()) {
            log.debug("本地商品列表为空，无需比较价格");
            return Collections.emptyList();
        }

        if (jdGoodsMap == null || jdGoodsMap.isEmpty()) {
            log.debug("京东商品映射为空，无需比较价格");
            return Collections.emptyList();
        }

        log.debug("开始比较价格差异，本地商品: {}, 京东商品: {}, 阈值: {}", Integer.valueOf(localGoods.size()), Integer.valueOf(jdGoodsMap.size()), threshold);

        List<PriceChangeDto> priceChanges = new ArrayList<>();
        int comparedCount = 0;
        int notFoundCount = 0;
        int noChangeCount = 0;
        int belowThresholdCount = 0;

        for (Goods localGood : localGoods) {
            comparedCount++;
            Goods jdGood = jdGoodsMap.get(localGood.getGoodsId());
            if (jdGood == null) {
                notFoundCount++;
                log.debug("商品ID {} 在京东数据中未找到", localGood.getGoodsId());
                continue;
            }

            // 比较价格
            if (!isPriceChanged(localGood.getPrice(), jdGood.getPrice())) {
                noChangeCount++;
                continue;
            }

            PriceChangeDto priceChange = PriceChangeDto.create(
                localGood.getGoodsId(),
                localGood.getGoodsName(),
                localGood.getPrice(),
                jdGood.getPrice()
            );

            // 检查是否超过变更阈值
            if (threshold != null && !priceChange.isMajorChange(threshold)) {
                belowThresholdCount++;
                log.debug("价格变更未超过阈值，跳过: 商品ID={}, 变更金额={}, 阈值={}",
                         localGood.getGoodsId(), priceChange.getChangeAmount(), threshold);
                continue;
            }

            priceChanges.add(priceChange);
            log.debug("发现价格变更: 商品ID={}, {}",
                     localGood.getGoodsId(), priceChange.getChangeDescription());
        }

        log.info("价格比较完成: 比较={}, 未找到={}, 无变化={}, 低于阈值={}, 有效变更={}",
                Integer.valueOf(comparedCount), Integer.valueOf(notFoundCount), Integer.valueOf(noChangeCount),
                Integer.valueOf(belowThresholdCount), Integer.valueOf(priceChanges.size()));

        return priceChanges;
    }



    @Override
    public Map<Long, Goods> getJdGoodsMap(List<Long> goodsIds) {
        Map<Long, Goods> jdGoodsMap = new HashMap<>();

        if (goodsIds == null || goodsIds.isEmpty()) {
            log.warn("商品ID列表为空，无法获取京东商品信息");
            return jdGoodsMap;
        }

        log.info("开始获取 {} 个商品的京东详情信息", Integer.valueOf(goodsIds.size()));

        int successCount = 0;
        int failedCount = 0;

        // 逐个获取商品详情信息
        for (Long goodsId : goodsIds) {
            try {
                log.debug("获取商品 {} 的详情信息", goodsId);

                // 调用京东API获取商品详情
                GoodsDetailInfoResponseDto detailResponse = jdCategorySyncService.getGoodsDetailInfo(String.valueOf(goodsId), null, null);

                if (detailResponse != null && detailResponse.getData() != null) {
                    // 将详情信息转换为Goods对象
                    Goods goods = convertDetailToGoods(detailResponse.getData());
                    if (goods != null) {
                        jdGoodsMap.put(goodsId, goods);
                        successCount++;
                        log.debug("成功获取商品 {} 的详情信息，价格: {}", goodsId, goods.getPrice());
                    } else {
                        log.warn("商品 {} 详情信息转换失败", goodsId);
                        failedCount++;
                    }
                } else {
                    log.warn("商品 {} 详情信息为空", goodsId);
                    failedCount++;
                }

                // 添加短暂延迟，避免对京东API造成过大压力
                Thread.sleep(100);

            } catch (IOException e) {
                log.error("获取商品 {} 详情信息时发生IO异常: {}", goodsId, e.getMessage());
                failedCount++;
            } catch (InterruptedException e) {
                log.error("获取商品详情信息时被中断: {}", e.getMessage());
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                log.error("获取商品 {} 详情信息时发生未知异常: {}", goodsId, e.getMessage(), e);
                failedCount++;
            }
        }

        log.info("京东商品详情获取完成: 成功={}, 失败={}, 总数={}", Integer.valueOf(successCount), Integer.valueOf(failedCount), Integer.valueOf(goodsIds.size()));
        return jdGoodsMap;
    }

    /**
     * 将商品详情DTO转换为Goods对象
     *
     * @param detailDto 商品详情DTO
     * @return Goods对象，如果转换失败返回null
     */
    private Goods convertDetailToGoods(GoodsDetailDto detailDto) {
        if (detailDto == null) {
            return null;
        }

        try {
            Goods goods = new Goods();

            // 设置基本信息
            goods.setGoodsId(detailDto.getGoodsId());
            goods.setGoodsName(detailDto.getGoodsName());

            // 设置价格信息（这是最重要的）
            if (detailDto.getPrice() != null && !detailDto.getPrice().isEmpty()) {
                try {
                    goods.setPrice(new BigDecimal(detailDto.getPrice()));
                } catch (NumberFormatException e) {
                    log.warn("商品 {} 价格格式错误: {}", detailDto.getGoodsId(), detailDto.getPrice());
                    return null;
                }
            } else {
                log.warn("商品 {} 价格信息为空", detailDto.getGoodsId());
                return null;
            }

            // 设置其他信息
            goods.setGoodsState(detailDto.getGoodsState());

            // 设置市场价
            if (detailDto.getMarketPrice() != null && !detailDto.getMarketPrice().isEmpty()) {
                try {
                    goods.setMarketPrice(new BigDecimal(detailDto.getMarketPrice()));
                } catch (NumberFormatException e) {
                    log.debug("商品 {} 市场价格式错误: {}", detailDto.getGoodsId(), detailDto.getMarketPrice());
                    // 市场价格式错误不影响主要功能，继续处理
                }
            }
            // 设置成本价
            if (detailDto.getCostPrice() != null) {
                goods.setCostPrice(detailDto.getCostPrice());
            }
            return goods;

        } catch (Exception e) {
            log.error("转换商品详情时发生异常，商品ID: {}, 错误: {}", detailDto.getGoodsId(), e.getMessage(), e);
            return null;
        }
    }


    /**
     * 判断价格是否发生变更
     */
    private boolean isPriceChanged(BigDecimal oldPrice, BigDecimal newPrice) {
        if (oldPrice == null && newPrice == null) {
            return false;
        }
        if (oldPrice == null || newPrice == null) {
            return true;
        }
        return oldPrice.compareTo(newPrice) != 0;
    }

    /**
     * 创建同步日志
     */
    private List<PriceSyncLog> createSyncLogs(List<PriceChangeDto> priceChanges, String syncBatchId) {
        return priceChanges.stream().map(change -> {
            PriceSyncLog log = new PriceSyncLog();
            log.setSyncBatchId(syncBatchId);
            log.setGoodsId(change.getGoodsId());
            log.setGoodsName(change.getGoodsName());
            log.setOldPrice(change.getOldPrice());
            log.setNewPrice(change.getNewPrice());
            log.setChangeAmount(change.getChangeAmount());
            log.setChangePercentage(change.getChangePercentage());
            log.setSyncStatus(Integer.valueOf(PriceSyncLog.SyncStatus.PENDING));
            return log;
        }).collect(Collectors.toList());
    }

    @Override
    public RetryResult retryFailedSync(int maxRetryCount) {
        log.info("开始重试失败的同步记录，最大重试次数: {}", Integer.valueOf(maxRetryCount));

        List<PriceSyncLog> failedLogs = priceSyncLogRepository.findFailedLogsForRetry(Integer.valueOf(maxRetryCount));
        if (failedLogs.isEmpty()) {
            log.info("没有需要重试的失败记录");
            return RetryResult.noRetry();
        }

        log.info("找到 {} 条需要重试的失败记录", Integer.valueOf(failedLogs.size()));

        // 按商品ID分组，构建价格变更列表
        List<PriceChangeDto> retryChanges = failedLogs.stream()
                .map(log -> PriceChangeDto.create(log.getGoodsId(), log.getGoodsName(),
                                                 log.getOldPrice(), log.getNewPrice()))
                .collect(Collectors.toList());

        // 重新推送
        String retryBatchId = generateSyncBatchId() + "_RETRY";
        PushResult pushResult = pushPriceChanges(retryChanges, retryBatchId);

        // 更新本地价格
        if (pushResult.getSuccessCount() > 0) {
            updateLocalPrices(pushResult.getSuccessChanges());
        }

        // 根据推送结果创建重试结果
        RetryResult result;
        if (pushResult.getFailedCount() == 0) {
            result = RetryResult.success(failedLogs.size());
        } else if (pushResult.getSuccessCount() == 0) {
            result = RetryResult.failure(failedLogs.size(), "重试全部失败");
        } else {
            result = RetryResult.partial(failedLogs.size(), pushResult.getSuccessCount(), pushResult.getFailedCount());
        }
        result.markEnd();

        log.info("重试完成: {}", result);
        return result;
    }

    /**
     * 创建空的同步结果
     */
    private PriceSyncResult createEmptyResult(String syncBatchId, long executionTime) {
        return PriceSyncResult.noChange(syncBatchId, 0, 0, executionTime);
    }

    /**
     * 记录同步失败
     */
    private void recordSyncFailure(String syncBatchId, String errorMessage) {
        try {
            // 这里可以记录到专门的失败日志表或发送告警
            log.error("同步失败记录 - 批次ID: {}, 错误: {}", syncBatchId, errorMessage);
        } catch (Exception e) {
            log.error("记录同步失败信息时发生错误", e);
        }
    }

    /**
     * 生成同步批次ID
     */
    private String generateSyncBatchId() {
        return SYNC_BATCH_PREFIX + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMddHHmmss"));
    }

    @Override
    public PushResult pushPriceChanges(List<PriceChangeDto> priceChanges, String syncBatchId) {
        if (priceChanges == null || priceChanges.isEmpty()) {
            log.info("没有需要推送的价格变更");
            return PushResult.empty().markEnd();
        }

        List<PriceChangeDto> successChanges = new ArrayList<>();
        List<PriceChangeDto> failedChanges = new ArrayList<>();

        log.info("开始推送价格变更，总数量: {}", Integer.valueOf(priceChanges.size()));

        // 逐个推送商品价格
        for (PriceChangeDto priceChange : priceChanges) {
            try {
                // 调用第三方接口更新单个商品价格
                RBase response = mallGoodsFeignApi.goodsUpdatePrice(
                    String.valueOf(priceChange.getGoodsId()),
                    priceChange.getOldPrice(),  // 原价
                    priceChange.getNewPrice()   // 售价
                );

                if (response != null && response.getStatus() == 200) {
                    // 推送成功
                    successChanges.add(priceChange);
                    updateSingleSyncLogStatus(syncBatchId, priceChange, Integer.valueOf(PriceSyncLog.SyncStatus.SUCCESS),
                                            "推送成功: " + response.getMsg());
                    log.debug("商品 {} 价格推送成功", priceChange.getGoodsId());
                } else {
                    // 推送失败
                    failedChanges.add(priceChange);
                    String errorMsg = response != null ? response.getMsg() : "响应为空";
                    updateSingleSyncLogStatus(syncBatchId, priceChange, Integer.valueOf(PriceSyncLog.SyncStatus.FAILED), "推送失败: " + errorMsg);
                    log.warn("商品 {} 价格推送失败: {}", priceChange.getGoodsId(), errorMsg);
                }

            } catch (Exception e) {
                // 推送异常
                failedChanges.add(priceChange);
                updateSingleSyncLogStatus(syncBatchId, priceChange, Integer.valueOf(PriceSyncLog.SyncStatus.FAILED),
                                        "推送异常: " + e.getMessage());
                log.error("商品 {} 价格推送异常", priceChange.getGoodsId(), e);
            }
        }

        log.info("价格推送完成: 成功={}, 失败={}", Integer.valueOf(successChanges.size()), Integer.valueOf(failedChanges.size()));

        // 根据结果创建推送结果对象
        if (failedChanges.isEmpty() && !successChanges.isEmpty()) {
            return PushResult.success(successChanges).markEnd();
        } else if (successChanges.isEmpty() && !failedChanges.isEmpty()) {
            return PushResult.failure(failedChanges, "全部推送失败").markEnd();
        } else if (!successChanges.isEmpty() && !failedChanges.isEmpty()) {
            return PushResult.partial(successChanges, failedChanges, "部分推送失败").markEnd();
        } else {
            return PushResult.empty().markEnd();
        }
    }

    @Override
    @Transactional
    public void updateLocalPrices(List<PriceChangeDto> priceChanges) {
        if (priceChanges.isEmpty()) {
            return;
        }

        log.info("开始更新本地商品价格，数量: {}", Integer.valueOf(priceChanges.size()));

        for (PriceChangeDto change : priceChanges) {
            try {
                Goods goods = goodsRepository.findByGoodsId(change.getGoodsId());
                if (goods != null) {
                    goods.setPrice(change.getNewPrice());
                    goodsRepository.save(goods);
                    log.debug("更新商品价格成功: 商品ID={}, 新价格={}",
                             change.getGoodsId(), change.getNewPrice());
                } else {
                    log.warn("更新价格时未找到商品: {}", change.getGoodsId());
                }
            } catch (Exception e) {
                log.error("更新商品价格失败: 商品ID={}", change.getGoodsId(), e);
            }
        }

        log.info("本地商品价格更新完成");
    }

    /**
     * 更新同步日志状态
     */
    private void updateSyncLogStatus(String syncBatchId, List<PriceChangeDto> priceChanges,
                                   Integer status, String result) {
        List<Long> goodsIds = priceChanges.stream()
                .map(PriceChangeDto::getGoodsId)
                .collect(Collectors.toList());

        List<PriceSyncLog> logs = priceSyncLogRepository.findBySyncBatchId(syncBatchId);

        for (PriceSyncLog log : logs) {
            if (goodsIds.contains(log.getGoodsId())) {
                log.setSyncStatus(status);
                log.setSyncResult(result);
                log.setSyncTime(LocalDateTime.now());
                if (status == PriceSyncLog.SyncStatus.FAILED) {
                    log.setRetryCount(Integer.valueOf(log.getRetryCount() + 1));
                }
            }
        }

        priceSyncLogRepository.saveAll(logs);
    }

    /**
     * 更新单个商品的同步日志状态
     */
    private void updateSingleSyncLogStatus(String syncBatchId, PriceChangeDto priceChange,
                                         Integer status, String result) {
        try {
            List<PriceSyncLog> logs = priceSyncLogRepository.findBySyncBatchId(syncBatchId);

            for (PriceSyncLog log : logs) {
                if (log.getGoodsId().equals(priceChange.getGoodsId())) {
                    log.setSyncStatus(status);
                    log.setSyncResult(result);
                    log.setSyncTime(LocalDateTime.now());
                    if (status == PriceSyncLog.SyncStatus.FAILED) {
                        log.setRetryCount(Integer.valueOf(log.getRetryCount() + 1));
                    }
                    priceSyncLogRepository.save(log);
                    break;
                }
            }
        } catch (Exception e) {
            log.error("更新商品 {} 同步日志状态失败", priceChange.getGoodsId(), e);
        }
    }

}